﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.ServiceModel;
using Microsoft.WindowsAzure.ServiceRuntime;
using System.Reactive;
using System.Net;
using System.Reactive.Linq;
using System.Reactive.Disposables;
using Microsoft.WindowsAzure.StorageClient;
using System.Diagnostics.Contracts;
using RxCloud;

namespace Microsoft.WindowsAzure.StorageClient
{
    /// <summary>
    /// Extension methods for CloudQueue.
    /// </summary>
    public static class ServerExtensions
    {
        public const int PoisonMessageDequeueCount = 5;


        private static readonly RetryPolicy DefaultPolicy = RetryPolicies.RetryExponential(10, TimeSpan.FromMilliseconds(100));

        /// <summary>
        /// Converts CloudQueue to stream of messages.
        /// </summary>
        /// <param name="queue">Cloud queue</param>
        /// <param name="server">Notifications server</param>
        public static IObservable<CloudQueueMessage> ToObservable(this CloudQueue queue, NotificationsServer server)
        {
            Contract.Requires(queue != null);
            Contract.Requires(server != null);
            Contract.Ensures(Contract.Result<IObservable<CloudQueueMessage>>() != null);

            return queue.ToObservable(server, DefaultPolicy);
        }

        /// <summary>
        /// Converts CloudQueue to stream of messages.
        /// </summary>
        /// <param name="queue">Cloud queue</param>
        /// <param name="server">Notifications server</param>
        /// <param name="retryPolicy">Retry policy to calculate poll timeout.</param>
        public static IObservable<CloudQueueMessage> ToObservable(this CloudQueue queue, NotificationsServer server, RetryPolicy retryPolicy)
        {
            Contract.Requires(queue != null);
            Contract.Requires(server != null);
            Contract.Requires(retryPolicy != null);
            Contract.Ensures(Contract.Result<IObservable<CloudQueueMessage>>() != null);

            return queue.ToObservable(server.GetQueueNotification(queue.Name), retryPolicy);
        }

        /// <summary>
        /// Converts CloudQueue to stream of messages.
        /// </summary>
        /// <param name="queue">Cloud queue</param>
        /// <param name="haveMoreMessages">Notifications stream.</param>
        public static IObservable<CloudQueueMessage> ToObservable<T>(
                                                        this CloudQueue queue,
                                                        IObservable<T> haveMoreMessages)
        {
            Contract.Requires(queue != null);
            Contract.Requires(haveMoreMessages != null);
            Contract.Ensures(Contract.Result<IObservable<CloudQueueMessage>>() != null);

            return queue.ToObservable(haveMoreMessages, DefaultPolicy);
        }


        /// <summary>
        /// Converts CloudQueue to stream of messages.
        /// </summary>
        /// <param name="queue">Cloud queue.</param>
        /// <param name="haveMoreMessages">Notifications stream.</param>
        /// <param name="retryPolicy">Retry policy to calculate poll timeout.</param>
        public static IObservable<CloudQueueMessage> ToObservable<T>(
                                                        this CloudQueue queue,
                                                        IObservable<T> haveMoreMessages,
                                                        RetryPolicy retryPolicy)
        {
            Contract.Requires(queue != null);
            Contract.Requires(haveMoreMessages != null);
            Contract.Requires(retryPolicy != null);
            Contract.Ensures(Contract.Result<IObservable<CloudQueueMessage>>() != null);

            var iterator = Observable.Create<CloudQueueMessage>(
                                obs => Iterator(obs, queue, haveMoreMessages, retryPolicy));

            return Observable.Create<CloudQueueMessage>(
                obs => {
                    //Start message polling to process message after role start
                    IDisposable subscription = iterator.Subscribe(
                                                                obs.OnNext,
                                                                obs.OnError,
                                                                () => subscription = null);
                    return haveMoreMessages.Subscribe(
                    _ =>
                    {
                        if (subscription == null)
                        {
                            subscription = iterator.Subscribe(
                                                        obs.OnNext,
                                                        obs.OnError,
                                                        () => subscription = null);
                        }
                    },
                    () =>
                    {
                        if (subscription == null)
                        {
                            subscription.Dispose();
                        }
                    });
                });
        }

        private static IEnumerable<IObservable<object>> Iterator<T>(
                                                            IObserver<CloudQueueMessage> result,
                                                            CloudQueue queue,
                                                            IObservable<T> haveMoreMessages,
                                                            RetryPolicy retryPolicy)
        {
            Contract.Requires(result != null);
            Contract.Requires(queue != null);
            Contract.Requires(haveMoreMessages != null);
            Contract.Requires(retryPolicy != null);
            Contract.Ensures(Contract.Result<IEnumerable<IObservable<object>>>() != null);

            var shouldRetry = retryPolicy();
            var idleCount = 0;
            while (true)
            {
                var msgsObs = queue.GetMessagesAsync(32).ToListObservable();
                yield return msgsObs;
                var msgs = msgsObs[0];

                var hasMessages = false;
                foreach (var msg in msgs)
                {                    
                    if (msg.DequeueCount > PoisonMessageDequeueCount)
                    {
                        //Remove poison message
                        yield return queue.DeleteMessageAsync(msg).ToListObservable();
                    }
                    else
                    {
                        hasMessages = true;
                        idleCount = 0;

                        //Process message
                        result.OnNext(msg);
                    }
                }

                if (!hasMessages)
                {
                    TimeSpan delay;
                    if (!shouldRetry(idleCount++, null, out delay))
                    {
                        yield break;
                    }

                    //Wait timeout or notification
                    yield return Observable.Timer(delay)
                                           .TakeUntil(haveMoreMessages)
                                           .ToListObservable();
                }
            }

        }
    }
}
